package com.soft863.tc.consumer.copeocessor;

import com.soft863.tc.common.bean.BaseDao;
import com.soft863.tc.common.constant.Names;
import com.soft863.tc.common.constant.ValueConstant;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.coprocessor.RegionObserver;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.wal.WALEdit;


import java.io.IOException;
import java.util.Optional;

/*
 * 使用协处理器保存被叫用户的数据
 *
 * 协处理器的使用
 * 1.创建类
 * 2.让表知道协处理类（关联表）
 * 3.打包发布到hbase，关联的jar也需要发布，并且分发
 * */
public class InsertCalleeCoprocessor implements RegionObserver, RegionCoprocessor {
    private RegionCoprocessorEnvironment env;

    @Override
    public void start(CoprocessorEnvironment env) throws IOException {
        this.env = (RegionCoprocessorEnvironment) env;
    }

    @Override
    public Optional<RegionObserver> getRegionObserver() {
        return Optional.of(this);
    }
    //doPut ：模板方法设计模式
    //存在父子类，父类搭建算法的骨架，子类重写算法的细节

    /**
     * 保存主叫用户数据之后，由HBase自动保存被叫用户数据
     *
     * @param c
     * @param put
     * @param edit
     * @param durability
     * @throws IOException
     */
    @Override
    public void postPut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit, Durability durability)
            throws IOException {

        String callerRowKey = Bytes.toString(put.getRow());
        String[] values = callerRowKey.split("_");
        //1_156_2020-02_156_1345_1
        String calling = values[1];
        String called = values[3];
        String callTime = values[2];
        String duration = values[4];
        String flag = values[5];

        if ("1".equals(flag)) {
            //获取表
            Table table = env.getConnection().getTable(TableName.valueOf(ValueConstant.TABLE_NAME));
            String calleeRowKey = new CoprocessorDao()
                    .getRegionNum0(called, callTime) + "_" + called + "_" + callTime + "_" + calling + "_" + duration + "_0";
            Put put1 = new Put(Bytes.toBytes(calleeRowKey));
            byte[] family = Bytes.toBytes(Names.CF_CALLEE.getValue());
            put1.addColumn(family, Bytes.toBytes("call1"), Bytes.toBytes(called));
            put1.addColumn(family, Bytes.toBytes("call2"), Bytes.toBytes(calling));
            put1.addColumn(family, Bytes.toBytes("callTime"), Bytes.toBytes(callTime));
            put1.addColumn(family, Bytes.toBytes("duration"), Bytes.toBytes(duration));
            put1.addColumn(family, Bytes.toBytes("flag"), Bytes.toBytes("0"));
            //保存数据
            table.put(put1);
            //关闭表
            table.close();
        }


    }

    private static class CoprocessorDao extends BaseDao {
        public int getRegionNum0(String tel, String time) {
            return genRegionNum(tel, time);
        }
    }
}
